package study.zhangbo.lock.distributed.redis;

import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

/**
 * Created by zhangbo on 2020/10/28.
 */
@Slf4j
public class RedisLock {

    private String lockName;

    private AtomicBoolean atomicBoolean;

    private RedisLockUtils redisLockUtils;

    private String lockValue;

    private static final ConcurrentMap<String, Queue<Thread>> LOCKS = new ConcurrentHashMap<>();

    public RedisLock(String lockName, RedisLockUtils redisLockUtils) {
        this.lockName = lockName;
        this.redisLockUtils = redisLockUtils;
        this.lockValue = UUID.randomUUID().toString().replaceAll("-","");
        if (!LOCKS.containsKey(lockName)) {
            createLock(lockName, 1000);
        }
    }

    public RedisLock(String lockName, RedisLockUtils redisLockUtils, int queueSize) {
        this.lockName = lockName;
        this.redisLockUtils = redisLockUtils;
        this.lockValue = UUID.randomUUID().toString().replaceAll("-","");
        if (!LOCKS.containsKey(lockName)) {
            createLock(lockName, queueSize);
        }
    }

    /**
     * 获取休眠中的锁线程
     *
     * @return
     */
    public static List<String> lockDaemon() {
        List<String> holdLocks = new ArrayList<>();
        if (!LOCKS.isEmpty()) {
            Set<String> locks = LOCKS.keySet();
            for (String lock : locks) {
                if (!LOCKS.get(lock).isEmpty()) {
                    holdLocks.add(lock);
                }
            }

        }
        return holdLocks;
    }


    public static boolean awakenThread(String lockName) {
        if (!LOCKS.get(lockName).isEmpty()) {
            log.debug("守护线程开始唤醒锁：{}",lockName);
            LockSupport.unpark(LOCKS.get(lockName).peek());
            return true;
        }
        return false;
    }


    private synchronized void createLock(String lockName, int size) {
        if (!LOCKS.containsKey(lockName)) {
            LOCKS.put(lockName, new ArrayBlockingQueue<Thread>(size));
        }
    }

    /**
     * 获取头部锁线程对象
     *
     * @param lockName
     * @return
     */
    public static Thread peek(String lockName) {
        if (LOCKS.containsKey(lockName)) {
            Queue<Thread> threadQueue = LOCKS.get(lockName);
            if (null != threadQueue && !threadQueue.isEmpty()) {
                return threadQueue.peek();
            }
        }
        return null;
    }


    /**
     * 弹出一个锁线程对象
     *
     * @param lockName
     * @return
     */
    public static Thread poll(String lockName) {
        if (LOCKS.containsKey(lockName)) {
            Queue<Thread> threadQueue = LOCKS.get(lockName);
            if (null != threadQueue && !threadQueue.isEmpty()) {
                return threadQueue.poll();
            }
        }
        return null;
    }


    /**
     * 添加等待锁线程
     *
     * @param lockName 锁名称
     * @param thread   等待锁的线程对象
     * @return
     */
    public static boolean pull(String lockName, Thread thread) {
        if (LOCKS.containsKey(lockName)) {
            Queue<Thread> threadQueue = LOCKS.get(lockName);
            return threadQueue.offer(thread);
        }
        return false;
    }


    /**
     * 加锁/等待获取锁操作
     */
    public void lock() {
        /*1、校验是否有锁*/
        if (redisLockUtils.hasKey(this.lockName)) {
            /*有锁，则偿试获取锁操作*/
            tryLock();
        } else {
            /*1.2：如果无锁，则偿试获取锁*/
            boolean success = redisLockUtils.setNx(this.lockName, lockValue);
            if (success) {
                /*1.2.1：获取成功则：创建atomicBoolean对象，设置值为true并创建看门狗线程，给加锁线程续期*/
                createLockDog();
            } else {
                /*获取锁失败，则偿试获取锁*/
                tryLock();
            }


        }
    }

    /**
     * 当前获取锁失败，偿试获取锁
     */
    private void tryLock() {
        /*1.1：如果有锁，则将当前线程放入等锁队列，并调用LockSupport.park()方法*/
        LOCKS.get(this.lockName).offer(Thread.currentThread());
        LockSupport.park();
        boolean success = redisLockUtils.setNx(this.lockName, lockValue);
        if (success) {
            /*1.1.1：若被唤醒，则偿试获取锁，若成功获取到锁，则从队列中将当前线程poll掉*/
            pollLockThread();
        } else {
            /*1.1.2：若被唤醒，但未能获取到锁(则说明是别的节点获取走了锁)，则循环偿试获取锁，若成功获取到锁，则从当前线程中poll掉，同时创建atomicBoolean对象，设置值为true并创建看门狗线程，给加锁线程续期*/
            do {
                success = redisLockUtils.setNx(this.lockName, lockValue);
            } while (!success);
            pollLockThread();


        }
    }

    /**
     * 成功获取到锁，将获取到锁的线程弹出
     */
    private void pollLockThread() {
        Thread thread = LOCKS.get(this.lockName).peek();
        if (!thread.getName().equals(Thread.currentThread().getName())) {
            throw new RuntimeException("当前持索线程非队列头部线程！");
        }
        LOCKS.get(this.lockName).poll();
        createLockDog();
    }

    /**
     * 构建看门狗线程
     */
    private void createLockDog() {
        atomicBoolean = new AtomicBoolean(true);
        /*创建atomicBoolean对象，设置值为true并创建看门狗线程，给加锁线程续期*/
        Thread thread = new Thread(new RedisLockDog(this.lockName, atomicBoolean, redisLockUtils));
        thread.setDaemon(true);
        thread.setName("Lock_Dog_" + this.lockName);
        thread.start();
    }

    /**
     * 释放锁操作
     */
    public void unlock() {
        /*1、设置atomicBoolean值为false，看门狗线程终止*/
        String value = redisLockUtils.getValue(this.lockName);
        if(!lockValue.equals(value)) {
            throw new RuntimeException("非自身持有锁，不允许释放");
        }
        atomicBoolean.compareAndSet(true, false);
        redisLockUtils.delKey(this.lockName);
        /*2、调用peek方法获取当前锁头部线程，若不为null则调用LockSupport.unpark()方法唤醒*/
        if (LOCKS.get(this.lockName).isEmpty()) {
            LockSupport.unpark(LOCKS.get(this.lockName).peek());
        }
    }

}
